/*
*    Author : heiden deng(dengjianquan@beyondcent.com)
*    2017/07/11
*    version 0.0.9
*    consumer项目提供的接口函数实现
*/

#include "htsc_consumer_intf.h"

#include<string.h>
#include <map>
#include <string>
#include <vector>
#include <set>
#include<algorithm>

#include <grpc/support/log.h>
#include <grpc/support/alloc.h>
#include <grpc/support/sync.h>
#include <src/core/lib/support/spinlock.h>
#include "htsc_common/htsc_types.h"
#include "htsc_registry/url.h"
#include "htsc_common/htsc_grpc_properties_constants.h"
#include "htsc_common/htsc_grpc_properties_tools.h"
#include "htsc_common/htsc_grpc_utils.h"
#include "htsc_registry/registry_contants.h"
#include "htsc_registry/htsc_grpc_registy_intf.h"
#include "htsc_registry/registry_utils.h"

#include "htsc_grpc_consumer_utils.h"
#include "htsc_router.h"
#include "ConditionRouter.h"
#include "htsc_loadbalance.h"
#include "PickfirstLB.h"
#include "RoundRobinLB.h"
#include "WeightRoundRobinLB.h"
#include "FailoverUtils.h"
#include "RequestsControllerUtils.h"


//保存某服务名下的provider列表
static  std::map<std::string, std::vector<provider_t*> > g_valid_providers;
//保存某服务名下的router列表
static  std::map<std::string, std::vector<Router*> > g_valid_routers;
//保存请求某服务名的客户端,用于router中的过滤
static  std::map<std::string, std::vector<url_t*> > g_consumers;

//保存客户端对应的负载均衡策略,key值由server_name组成
static std::map<std::string, std::string> g_consumer_lbstragry;

/* Protects provider_queue */
static gpr_mu g_providers_mu;
/* Allow only one access provider queue at once */
static gpr_spinlock g_checker_providers_mu = GPR_SPINLOCK_STATIC_INITIALIZER;

/* Protects consumer_queue */
static gpr_mu g_consumer_mu;
/* Allow only one access consumer queue at once */
static gpr_spinlock g_checker_consumer_mu = GPR_SPINLOCK_STATIC_INITIALIZER;

static bool g_initialized = false;

static bool g_isRequestLbMode = false;//默认是连接负载均衡

static htsc_grpc_loadbalance* pfLB = new PickfirstLB();
static htsc_grpc_loadbalance* rrLB = new RoundRobinLB();
static htsc_grpc_loadbalance* wrrLB = new WeightRoundRobinLB();

static FailoverUtils failoverUtils;

static RequestsControllerUtils g_request_controller_utils;


#define GRPC_PROVIDERS_LIST_LOCK_START  if(gpr_spinlock_trylock(&g_checker_providers_mu)){ \
                                           gpr_mu_lock(&g_providers_mu);

#define GRPC_PROVIDERS_LIST_LOCK_END     gpr_mu_unlock(&g_providers_mu);\
									     gpr_spinlock_unlock(&g_checker_providers_mu);\
                                        }

#define GRPC_CONSUMERS_LIST_LOCK_START  if(gpr_spinlock_trylock(&g_checker_consumer_mu)){ \
                                           gpr_mu_lock(&g_consumer_mu);

#define GRPC_CONSUMERS_LIST_LOCK_END       gpr_mu_unlock(&g_consumer_mu);\
									       gpr_spinlock_unlock(&g_checker_consumer_mu);\
                                        }


void init_providers_list() {
	if (!g_initialized) {
		char buf[HTSC_GRPC_PROPERTY_KEY_MAX_LEN] = {0};
		int value = 0;
		gpr_mu_init(&g_providers_mu);
		gpr_mu_init(&g_consumer_mu);
		pfLB->setProviders(&g_valid_providers);
		rrLB->setProviders(&g_valid_providers);
		wrrLB->setProviders(&g_valid_providers);
		htsc_grpc_properties_init();
		if (0 == htsc_grpc_properties_get_value(HTSC_GRPC_PROPERTIES_C_CONSUMER_SWITH_THRES, NULL, buf)) {
			value = atoi(buf);
			if (value > 0)
			{
				failoverUtils.setSwitchThreshold(value);

			}
			memset(buf, 0, HTSC_GRPC_PROPERTY_KEY_MAX_LEN);
		}
		if (0 == htsc_grpc_properties_get_value(HTSC_GRPC_PROPERTIES_C_CONSUMER_PUNISH_TIME, NULL, buf))
		{
			value = atoi(buf);
			if (value > 0)
			{
				failoverUtils.setPunishTime(value * 1000);
			}
		}
		memset(buf, 0, HTSC_GRPC_PROPERTY_KEY_MAX_LEN);
		if (0 == htsc_grpc_properties_get_value(HTSC_GRPC_PROPERTIES_C_LB_MODE, NULL, buf))
		{
			if (0 == strcmp(buf, HTSC_GRPC_LB_MODE_REQUEST)) {
				g_isRequestLbMode = true;
			}
			else {
				g_isRequestLbMode = false;
			}
			
		}

		g_initialized = true;
	}
}

bool is_request_loadbalance() {
	return g_isRequestLbMode;
}

char* htsc_grpc_get_sn_from_target(char* target) {
	if (!target)
	{
		return NULL;
	}
	std::string tgt = target;
	std::string::size_type startpos = tgt.find("///");
	if (startpos == std::string::npos)
	{
		return NULL;
	}
	startpos += 3;
	std::string::size_type endpos = tgt.find("?",startpos);
	if (endpos == std::string::npos)
	{
		endpos = tgt.length();
	}
	std::string sn = tgt.substr(startpos, endpos - startpos);
	return gprc_strdup(sn.c_str());
}


void revoker_providers_list_process(const char *sn) {
	if (!sn)
	{
		return;
	}
	std::map<std::string, std::vector<url_t*> >::iterator consumerIter = g_consumers.find(sn);
	if (consumerIter == g_consumers.end() || consumerIter->second.size() == 0)
	{
		return;
	}
	std::map<std::string, std::vector<Router*> >::iterator routeMapIter = g_valid_routers.find(sn);
	std::map<std::string, std::vector<provider_t*> >::iterator provider_lst_iter = g_valid_providers.find(sn);
	if (routeMapIter != g_valid_routers.end())
	{
		for (std::vector<Router*>::iterator routeIter = routeMapIter->second.begin();
			routeIter != routeMapIter->second.end();routeIter++)
		{
			Router *route = *routeIter;
			route->route(provider_lst_iter->second, consumerIter->second[0]);
		}
	}
}

bool provider_weight_comp(provider_t* p1, provider_t* p2) {
	if (!p1 || !p2)
	{
		return false;
	}
	if (p1->weight > p2->weight)
	{
		return true;
	}
	return false;
}

//更新缓存，首先清除原空间，再保存
void updateProvidersCache(const char* service_name, url_t* urls, int url_num) {
	GRPC_PROVIDERS_LIST_LOCK_START
	char provider_add[HOST_MAX_LEN];
	bool need_free_cache = false;
	std::set<std::string> new_providers;
	std::set<std::string> old_providers;
	std::map<std::string, std::vector<provider_t*> >::iterator provider_lst_iter = g_valid_providers.find(service_name);


	if (provider_lst_iter != g_valid_providers.end())
	{
		for (size_t i = 0; i < provider_lst_iter->second.size(); i++)
		{
			memset(provider_add, 0, HOST_MAX_LEN);
			sprintf(provider_add, "%s:%d", provider_lst_iter->second[i]->host, provider_lst_iter->second[i]->port);
			old_providers.insert(std::string(provider_add));
		}

		for (size_t i = 0; i < url_num; i++)
		{
			memset(provider_add, 0, HOST_MAX_LEN);
			sprintf(provider_add, "%s:%d", urls[i].host, urls[i].port);
			new_providers.insert(std::string(provider_add));
		}
		std::set<std::string>::iterator new_iter;
		for (std::set<std::string>::iterator old_iter = old_providers.begin(); 
			old_iter != old_providers.end(); old_iter++)
		{
			if (new_providers.find(*old_iter) == new_providers.end())
			{
				need_free_cache = true;
				break;
			}
		}
		if (need_free_cache)
		{
			std::vector<provider_t*>::iterator provider_iter = provider_lst_iter->second.begin();
			while (provider_iter != provider_lst_iter->second.end())
			{
				free_provider_v2((provider_t *)*provider_iter);
				provider_iter = provider_lst_iter->second.erase(provider_iter);
			}
			provider_lst_iter->second.clear();
			g_valid_providers.erase(provider_lst_iter);

			std::vector<provider_t*> providers;
			for (size_t i = 0; i < url_num; i++)
			{
				providers.push_back(provider_from_url(&urls[i]));
			}
			sort(providers.begin(), providers.end(), provider_weight_comp);//根据weight进行排序,weight值较大的排在前面
			g_valid_providers.insert(std::pair<std::string, std::vector<provider_t*> >(std::string(service_name), providers));
		}
		else {
			//todo添加新上线的服务
			for (size_t i = 0; i < url_num; i++)
			{
				memset(provider_add, 0, HOST_MAX_LEN);
				sprintf(provider_add, "%s:%d", urls[i].host, urls[i].port);
				if (old_providers.find(provider_add) == old_providers.end())//新增provider
				{
					provider_t *provider = provider_from_url(&urls[i]);
					HTSC_GRPC_SET_BIT(provider->flag, HTSC_GRPC_PROVIDER_FLAG_NOT_ACCESSED_BY_OLD_CLIENT);
					provider_lst_iter->second.push_back(provider);
				}
			}
		}
	}
	else {
		//todo直接插入新记录
		std::vector<provider_t*> providers;
		for (size_t i = 0; i < url_num; i++)
		{
			providers.push_back(provider_from_url(&urls[i]));
		}
		sort(providers.begin(), providers.end(), provider_weight_comp);//根据weight进行排序,weight值较大的排在前面
		g_valid_providers.insert(std::pair<std::string, std::vector<provider_t*> >(std::string(service_name), providers));
	}
	
	rrLB->resetCursor(service_name);
	wrrLB->resetCursor(service_name);
	//g_valid_providers.insert(std::pair<std::string, std::vector<provider_t*> >(std::string(service_name), providers));
	GRPC_PROVIDERS_LIST_LOCK_END
}


//消费者 providers目录订阅函数
void consumer_providers_callback(url_t *urls, int url_num) {
	char *service_name = url_get_parameter_v2(&urls[0], HTSC_GRPC_REGISTRY_KEY_INTERFACE, NULL);
	if (0 == strcmp(urls[0].protocol, HTSC_GRPC_EMPTY_PROTOCOL))
	{
		//todo
	}
	else {
		updateProvidersCache(service_name, urls, url_num);
	}

}

bool route_comp(Router* a, Router *b) {
	if (((ConditionRouter*)a)->get_priority() > ((ConditionRouter*)b)->get_priority())
	{
		return true;
	}
	return false;
}

//消费者routers目录订阅函数
void consumer_routers_callback(url_t *urls, int url_num) {
	if (!urls)
	{
		return;
	}
	bool isEmptyProtocol = false;
	if (0 == strcmp(urls[0].protocol, HTSC_GRPC_EMPTY_PROTOCOL))
	{
		isEmptyProtocol = true;;
	}
	std::map<std::string, std::vector<Router*> >::iterator routeMapIter = g_valid_routers.find(urls[0].path);
	//先清空原有规则
	if (routeMapIter != g_valid_routers.end())
	{
		std::vector<Router*> &routes = routeMapIter->second;
		std::vector<Router*>::iterator routerVecIter;
		for (routerVecIter = routes.begin(); routerVecIter != routes.end();)
		{
			delete *routerVecIter;
			routerVecIter = routes.erase(routerVecIter);
		}
	}
	else {
		std::vector<Router*> routes;
		g_valid_routers.insert(std::pair<std::string, std::vector<Router*> >(urls[0].path, routes));
	}
	routeMapIter = g_valid_routers.find(urls[0].path);
	if (!isEmptyProtocol)
	{
		for (size_t i = 0; i < url_num; i++)
		{
			Router *route = new ConditionRouter(urls + i);
			routeMapIter->second.push_back(route);
		}
		sort(routeMapIter->second.begin(), routeMapIter->second.end(), route_comp);
	}
	//当黑白名单有变化时，需要重置provider列表中的标志位
	GRPC_PROVIDERS_LIST_LOCK_START
	std::map<std::string, std::vector<provider_t*> >::iterator provider_lst_iter = g_valid_providers.find(urls[0].path);
	if (provider_lst_iter != g_valid_providers.end())
	{
		for (size_t i = 0; i < provider_lst_iter->second.size(); i++)
		{
			HTSC_GRPC_CLR_BIT(provider_lst_iter->second[i]->flag, HTSC_GRPC_PROVIDER_FLAG_IN_BLKLIST);
		}

	}
	GRPC_PROVIDERS_LIST_LOCK_END
	//黑白名单有更新，更新provider列表,设置provider黑名单标记
	revoker_providers_list_process(urls[0].path);
}

bool url_comp(url_t* a, url_t* b) {
	if (!a || !b)
	{
		return true;
	}
	if (strcmp(a->host,b->host) <= 0 )
	{
		return true;
	}
	return false;

}

//消费者 configurator订阅函数
void consumer_configurators_callback(url_t *urls, int url_num) {
	if (0 == url_num || 0 == strcmp(urls[0].protocol, HTSC_GRPC_EMPTY_PROTOCOL))
	{
		return;
	}
	char *service_name = url_get_parameter_v2(&urls[0], HTSC_GRPC_REGISTRY_KEY_INTERFACE, NULL);
	std::vector<url_t*> urlVec;
	for (size_t i = 0; i < url_num; i++)
	{
		urlVec.push_back(urls + i);
	}
	sort(urlVec.begin(), urlVec.end(), url_comp);
	char *ip = NULL;
	char *param = NULL;
	int weight = 0;

	GRPC_PROVIDERS_LIST_LOCK_START

	std::map<std::string, std::vector<provider_t*> >::iterator provider_lst_iter = g_valid_providers.find(service_name);
	for (size_t i = 0; i < urlVec.size(); i++)
	{
		ip = urlVec[i]->host;
		if (provider_lst_iter != g_valid_providers.end())
		{
			for (size_t j = 0;j < provider_lst_iter->second.size(); j++) {
				if ((0 == strcmp(ip, HTSC_GRPC_ANYHOST_VALUE)) || 
					(0 == strcmp(ip, provider_lst_iter->second[j]->host)))
				{
					param = url_get_parameter_v2(urlVec[i], HTSC_GRPC_REGISTRY_KEY_WEIGHT, NULL);
					if (param)
					{
						weight = atoi(param);
						if (weight >0 )
						{
							provider_lst_iter->second[j]->weight = weight;
						}
					}
					param = url_get_parameter_v2(urlVec[i], HTSC_GRPC_REGISTRY_KEY_DEPRECATED, NULL);
					if (param)
					{
						if (0 == strcmp(param,"true")) {
							provider_lst_iter->second[j]->deprecated = true;
						}
						else {
							provider_lst_iter->second[j]->deprecated = false;
						}
					}
				}
			}
		}
		//todo 处理客户端最大连接数参数设置

	}
	GRPC_PROVIDERS_LIST_LOCK_END

	//更新客户端负载均衡策略配置以及流量控制参数
	char *lb = NULL;
	std::map<std::string, std::string>::iterator lbIter;
	for (size_t i = 0; i < urlVec.size(); i++) {
		if (0 == strcmp(urlVec[i]->host, HTSC_GRPC_ANYHOST_VALUE) ||
			0 == strcmp(urlVec[i]->host, get_local_ip()))
		{
			lb = url_get_parameter_v2(urlVec[i], HTSC_GRPC_REGISTRY_KEY_DEFAULT_LOADBALANCE, NULL);
			if (lb)
			{
				lbIter = g_consumer_lbstragry.find(urlVec[i]->path);
				if (lbIter == g_consumer_lbstragry.end())
				{
					g_consumer_lbstragry.insert(std::pair<std::string, std::string>(urlVec[i]->path, lb));
				}
				else {
					lbIter->second = std::string(lb);
				}
			}
			lb = url_get_parameter_v2(urlVec[i], HTSC_GRPC_CONSUMER_DEFAULT_REQUEST, NULL);
			if (lb)
			{
				g_request_controller_utils.SetMaxRequestMap(urlVec[i]->path, atol(lb));
			}
		}
		
	}
	revoker_providers_list_process(urls[0].path);
}

provider_t* get_providers_intf(const char* service_name) {
	if (!service_name)
	{
		return NULL;
	}
	std::map<std::string, std::string>::iterator lbIter = g_consumer_lbstragry.find(service_name);
	std::string strLbStragry = HTSC_GRPC_DEFAULT_LB_TYPE;
	if (lbIter != g_consumer_lbstragry.end())
	{
		strLbStragry = lbIter->second;
	}
	if (0 == strcmp(strLbStragry.c_str(), HTSC_GRPC_LB_TYPE_PF))
	{
		return pfLB->chooseProvider(service_name);
	}
	else if (0 == strcmp(strLbStragry.c_str(), HTSC_GRPC_LB_TYPE_WRR)) {
		return wrrLB->chooseProvider(service_name);
	}
	else {
		return rrLB->chooseProvider(service_name);
	}

}



void get_all_providers_by_name(const char*service_name) {
	char query_str[HTSC_GRPC_PATH_MAX_LEN] = { 0 };
	size_t i = 0;
	if (!service_name)
	{
		return ;
	}

	int provider_nums = 0;

	strcat(query_str, HTSC_GRPC_REGISTRY_KEY_GRPC);
	strcat(query_str, "://0.0.0.0/");
	strcat(query_str, service_name);

	strcat(query_str, "?");
	strcat(query_str, HTSC_GRPC_CATEGORY_KEY);
	strcat(query_str, "=");
	strcat(query_str, HTSC_GRPC_PROVIDERS_CATEGORY);

	strcat(query_str, "&");
	strcat(query_str, HTSC_GRPC_REGISTRY_KEY_INTERFACE);
	strcat(query_str, "=");
	strcat(query_str, service_name);

	url_t *query_url = url_parse(query_str);

	url_t* provider_urls = lookup(query_url, &provider_nums);
	if (provider_urls)
	{
		updateProvidersCache(service_name, provider_urls, provider_nums);
		for (size_t i = 0; i < provider_nums; i++)
		{
			url_free(provider_urls + i);
		}
		free(provider_urls);
	}
	url_full_free(&query_url);
	revoker_providers_list_process(service_name);
}

provider_t* consumer_query_providers(const char *service_name, int *nums) {
	char query_str[HTSC_GRPC_PATH_MAX_LEN] = {0};
	size_t i = 0;
	if (!service_name)
	{
		return NULL;
	}
	
	int provider_nums = 0;
	provider_t *providers = NULL;
	*nums = 0;
#ifdef DEBUG_DIRECT_QUERY
	strcat(query_str, HTSC_GRPC_REGISTRY_KEY_GRPC);
	strcat(query_str, "://0.0.0.0/");
	strcat(query_str, service_name);

	strcat(query_str, "?");
	strcat(query_str, HTSC_GRPC_CATEGORY_KEY);
	strcat(query_str, "=");
	strcat(query_str, HTSC_GRPC_PROVIDERS_CATEGORY);

	strcat(query_str, "&");
	strcat(query_str, HTSC_GRPC_REGISTRY_KEY_INTERFACE);
	strcat(query_str, "=");
	strcat(query_str, service_name);

	url_t *query_url = url_parse(query_str);

	url_t* provider_urls = lookup(query_url, &provider_nums);

	*nums = provider_nums;
	if (nums > 0)
	{
		providers = (provider_t*)gpr_zalloc(sizeof(provider_t) * provider_nums);
		for (i = 0; i < provider_nums; i++)
		{
			init_provider_from_url(providers + i, provider_urls + i);
		}
	}
	url_full_free(&query_url);
	for (i = 0; i < provider_nums; i++)
	{
		url_free(provider_urls + i);
	}
	free(provider_urls);
#else
	
	GRPC_PROVIDERS_LIST_LOCK_START
	providers = get_providers_intf(service_name);
	if (providers)
	{
		*nums = 1;
	}
	/*原始代码代码，返回所有providers，未应用负载均衡策略
	std::map<std::string, std::vector<provider_t*> >::iterator provider_lst_iter = g_valid_providers.find(service_name);
	if (provider_lst_iter != g_valid_providers.end())
	{
		//首先统计合法的provider个数
		for (std::vector<provider_t*>::iterator providerIter = provider_lst_iter->second.begin();
			providerIter != provider_lst_iter->second.end(); providerIter++)
		{
			provider_t* provider = *providerIter;
			if (!HTSC_GRPC_CHECK_BIT(provider->flag, HTSC_GRPC_PROVIDER_FLAG_IN_BLKLIST))
			{
				provider_nums++;
			}
		}
		if (provider_nums != 0)
		{
			providers = (provider_t*)gpr_zalloc(sizeof(provider_t) * provider_nums);
			i = 0;
			for (std::vector<provider_t*>::iterator providerIter = provider_lst_iter->second.begin();
				providerIter != provider_lst_iter->second.end(); providerIter++)
			{
				provider_t* provider = *providerIter;
				if (!HTSC_GRPC_CHECK_BIT(provider->flag, HTSC_GRPC_PROVIDER_FLAG_IN_BLKLIST))
				{
					strcpy(providers[i].host, provider->host);
					providers[i].port = provider->port;
					i++;
				}
			}
			*nums = provider_nums;
		}
		
	}
	*/

	GRPC_PROVIDERS_LIST_LOCK_END
	
#endif
	return providers;
}

//consumer注册。
//1、解析fullmethod 获取conf文件配置信息、拼接consumer url串，调用接口写入向zk写入url串、
// 记录已经注册的url串，以便consumer停止时调用zk接口注销url串。
//2、注册监听器
//  a.消费者订阅providers目录 
//  b.消费者订阅routers目录
//  c.消费者订阅configurators目录
char * htsc_grpc_consumer_register(const char *fullmethod)
{
	init_providers_list();
	//未加锁
	char *consumerurl = htsc_grpc_consumer_url(fullmethod);
	//printf("-------- register url .... \n %s ", consumerurl);
	url_t *url = url_parse(consumerurl);
	registry(url);

	//缓存该客户端信息
	GRPC_CONSUMERS_LIST_LOCK_START
	std::map<std::string, std::vector<url_t*> >::iterator consumerIter = g_consumers.find(fullmethod);
	url_t *consumerUrl = (url_t*)gpr_zalloc(sizeof(url_t));
	url_init(consumerUrl);
	url_clone(url, consumerUrl);
	if (consumerIter == g_consumers.end())
	{	
		std::vector<url_t*> consumers;
		consumers.push_back(consumerUrl);
		g_consumers.insert(std::pair<std::string, std::vector<url_t*> >(fullmethod, consumers));
	}
	else {
		consumerIter->second.push_back(consumerUrl);
	}
	GRPC_CONSUMERS_LIST_LOCK_END

	//更新loadbalance配置
	std::map<std::string, std::string>::iterator lbIter = g_consumer_lbstragry.find(fullmethod);
	if (lbIter == g_consumer_lbstragry.end())
	{
		g_consumer_lbstragry.insert(std::pair<std::string, std::string>(fullmethod, HTSC_GRPC_LB_TYPE_RR));
		lbIter = g_consumer_lbstragry.find(fullmethod);
	}
	char *consumerLb = url_get_parameter_v2(url, HTSC_GRPC_REGISTRY_KEY_DEFAULT_LOADBALANCE, NULL);
	if (consumerLb)
	{
		lbIter->second = std::string(consumerLb);
	}

	//printf("-------- register listener ..... \n");

	//消费者订阅providers目录
	url_update_parameter(url, HTSC_GRPC_CATEGORY_KEY, HTSC_GRPC_PROVIDERS_CATEGORY);
	subscribe(url, consumer_providers_callback);


	//消费者订阅routers目录	
	url_update_parameter(url, HTSC_GRPC_CATEGORY_KEY, HTSC_GRPC_ROUTERS_CATEGORY);
	subscribe(url, consumer_routers_callback);


	//消费者订阅configurators目录
	url_update_parameter(url, HTSC_GRPC_CATEGORY_KEY, HTSC_GRPC_CONFIGURATORS_CATEGORY);
	subscribe(url, consumer_configurators_callback);

	url_full_free(&url);
	return consumerurl;
}

//consumer取消注册\取消监听器。
//1、注销zk注册信息。
//2、取消订阅监听器信息。
//  a.消费者取消订阅providers目录 
//  b.消费者取消订阅routers目录
//  c.消费者取消订阅configurators目录
int htsc_grpc_consumer_unregister(char * reginfo) {
	//取消zk注册
	//printf("-------- unregister .... \n");
	url_t *url = url_parse(reginfo);
	unregistry(url);
	

	//释放provider监听
	url_update_parameter(url, HTSC_GRPC_CATEGORY_KEY, HTSC_GRPC_PROVIDERS_CATEGORY);
	unsubscribe(url, consumer_providers_callback);

	//释放router监听
	url_update_parameter(url, HTSC_GRPC_CATEGORY_KEY, HTSC_GRPC_ROUTERS_CATEGORY);
	unsubscribe(url, consumer_routers_callback);

	//释放configurators监听
	url_update_parameter(url, HTSC_GRPC_CATEGORY_KEY, HTSC_GRPC_CONFIGURATORS_CATEGORY);
	unsubscribe(url, consumer_configurators_callback);

	//printf("-------- unregister finish: input any char for exit .... \n");
	//getchar();

	char *intf = url_get_parameter_v2(url, HTSC_GRPC_REGISTRY_KEY_INTERFACE, NULL);
	std::map<std::string, std::vector<provider_t*> >::iterator providerMapIter = g_valid_providers.find(intf);
	if (providerMapIter != g_valid_providers.end())
	{
		std::vector<provider_t*>::iterator providerIter;
		for (providerIter = providerMapIter->second.begin(); providerIter != providerMapIter->second.end();)
		{
			free_provider_v2_ex(*providerIter);
			providerIter = providerMapIter->second.erase(providerIter);
		}

	}

	std::map<std::string, std::vector<Router*> >::iterator routerMapIter = g_valid_routers.find(intf);
	if (routerMapIter != g_valid_routers.end())
	{
		std::vector<Router*>::iterator routerIter;
		for (routerIter = routerMapIter->second.begin(); routerIter != routerMapIter->second.end();)
		{
			delete *routerIter;
			routerIter = routerMapIter->second.erase(routerIter);
		}

	}

	std::map<std::string, std::vector<url_t*> >::iterator consumerIter = g_consumers.find(intf);
	if (consumerIter != g_consumers.end())
	{
		char buf[HTSC_GRPC_URL_MAX_LEN] = { 0 };
		std::vector<url_t*> &consumers = consumerIter->second;
		for (std::vector<url_t*>::iterator consumerIter = consumers.begin(); 
					consumerIter != consumers.end(); consumerIter++)
		{
			memset(buf, 0, HTSC_GRPC_URL_MAX_LEN);
			url_to_string_buf(*consumerIter, buf, HTSC_GRPC_URL_MAX_LEN);
			if (strcmp(buf,reginfo) == 0)
			{
				url_free(*consumerIter);
				consumers.erase(consumerIter);
				break;
			}
		}
	}

	url_full_free(&url);
	return 0;
}

void record_provider_failure(char* clientId, char* providerId) {
	failoverUtils.record_provider_failure(clientId, providerId);//使用示例对象调用静态方法
}

//获取经过黑白名单处理之后用于负载均衡的provider数量
int get_consumer_lb_providers_acount(char *service_name) {
	int count = 0;
	std::map<std::string, std::vector<provider_t*> >::iterator providerMapIter = g_valid_providers.find(service_name);
	if (providerMapIter != g_valid_providers.end())
	{
		for (size_t i = 0; i < providerMapIter->second.size(); i++)
		{
			if (!is_provider_invalid(providerMapIter->second[i])) {
				count++;
			}
		}
	}
	return count;
}

//获取zk上某服务的provider数量
int get_consumer_providers_acount(char* service_name) {
	int count = 0;
	std::map<std::string, std::vector<provider_t*> >::iterator providerMapIter = g_valid_providers.find(service_name);
	if (providerMapIter != g_valid_providers.end())
	{
		count = providerMapIter->second.size();
	}
	return count;
}

//设置某个provider 调用失败标记,isSet == 1表示设置标记位，否则为清除标记位
void set_or_clr_provider_failover_flag_inner(char* service_name, char *providerId,int isSet) {
	if (!service_name || !providerId)
	{
		return;
	}
	std::string provider = providerId;
	int pos = provider.find_first_of(':');
	if (pos == std::string::npos)
	{
		gpr_log(GPR_ERROR, "invalid providerId,[%s],should split by : ", providerId);
		return;
	}
	std::string provider_host = provider.substr(0, pos);
	int provider_port = 0;
	provider_port = atoi(provider.substr(pos + 1).c_str());

	GRPC_PROVIDERS_LIST_LOCK_START
	std::map<std::string, std::vector<provider_t*> >::iterator providerMapIter = g_valid_providers.find(service_name);	
	if (providerMapIter != g_valid_providers.end())
	{
		for (size_t i = 0; i < providerMapIter->second.size(); i++)
		{
			if (0 == strcmp(providerMapIter->second[i]->host, provider_host.c_str()) && 
				providerMapIter->second[i]->port == provider_port){
				if (isSet) {
					HTSC_GRPC_SET_BIT(providerMapIter->second[i]->flag, HTSC_GRPC_PROVIDER_FLAG_CALL_FAIlOVER);
				}
				else {
					HTSC_GRPC_CLR_BIT(providerMapIter->second[i]->flag, HTSC_GRPC_PROVIDER_FLAG_CALL_FAIlOVER);
				}
				break;
			}
		}
	}

	GRPC_PROVIDERS_LIST_LOCK_END
}

void set_provider_failover_flag(char* service_name, char *providerId) {
	set_or_clr_provider_failover_flag_inner(service_name, providerId, 1);
}

//清除某个provider 调用失败标记
void clr_provider_failover_flag(char* service_name, char *providerId) {
	set_or_clr_provider_failover_flag_inner(service_name, providerId, 0);
}

void set_consmuer_flow_control_threshold(char *service_name, long max_request) {
	if (!service_name || max_request <= 0)
	{
		return;
	}
	g_request_controller_utils.SetMaxRequestMap(service_name, max_request);
}

bool check_consumer_flow_control(char *service_name) {
	if (!service_name)
	{
		return true;
	}
	return g_request_controller_utils.CheckRequest(service_name);
}


void decrease_consumer_request_count(char *service_name) {
	if (!service_name)
	{
		return ;
	}
	g_request_controller_utils.DecreaseRequest(service_name);
}